package com.qys.livesMall.anchor.business;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.qys.livesMall.anchor.constants.RedisConstant;
import com.qys.livesMall.anchor.constants.enums.AnchorSellTaskStatus;
import com.qys.livesMall.anchor.controller.LiveStreamUtils;
import com.qys.livesMall.anchor.enums.AnchorMessageEnum;
import com.qys.livesMall.anchor.event.LiveEndEvent;
import com.qys.livesMall.anchor.event.pojo.LiveEndEventPOJO;
import com.qys.livesMall.anchor.mapper.H2AnchorMapper;
import com.qys.livesMall.anchor.model.dto.lives.sys.CreatePullFlowDTO;
import com.qys.livesMall.anchor.model.entity.*;
import com.qys.livesMall.anchor.model.vo.lives.LiveInfoVO;
import com.qys.livesMall.anchor.model.vo.lives.app.LivesStreamUrlVO;
import com.qys.livesMall.anchor.model.vo.lives.app.SettlementDataVO;
import com.qys.livesMall.anchor.service.*;
import com.qys.livesMall.app.user.model.entity.H2AppUser;
import com.qys.livesMall.app.user.utils.TencentCloudIm;
import com.qys.livesMall.cache.RedisCache;
import com.qys.livesMall.common.constants.Constants;
import com.qys.livesMall.common.exception.QysException;
import com.qys.livesMall.common.utils.MessageUtils;
import com.qys.livesMall.framework.utils.SecurityUtils;
import com.qys.livesMall.shop.mapper.ShopGoodsInfoMapper;
import com.qys.livesMall.shop.mapper.ShopGoodsSkuMapper;
import com.qys.livesMall.shop.mapper.ShopGoodsSkuStocksMapper;
import com.qys.livesMall.shop.modal.entity.ShopGoodsInfo;
import com.qys.livesMall.shop.modal.entity.ShopGoodsSku;
import com.qys.livesMall.shop.modal.entity.ShopGoodsSkuStocks;
import com.qys.livesMall.shop.modal.entity.ShopOrderSku;
import com.qys.livesMall.shop.modal.vo.task.app.child.SellTaskGoodsChild;
import com.qys.livesMall.shop.service.ShopGoodsInfoService;
import com.qys.livesMall.shop.service.ShopOrderSkuService;
import com.qys.livesMall.shop.service.ShopSellTaskProductService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class AnchorLivesBusiness {

    /**
     * 直播拉流错误码：被封禁用户
     */
    public static final String BANNED = "007";



    private final ShopGoodsInfoMapper shopGoodsInfoMapper;
    private final AnchorSellTaskService sellTaskService;
    private final AnchorSellLivesApplyService sellLivesApplyService;
    private final H2AnchorMapper anchorMapper;
    private final H2AnchorService anchorService;
    private final AnchorLivesRoomService livesRoomService;
    private final AnchorLivesSessionService livesSessionService;
    private final ShopSellTaskProductService shopSellTaskProductService;
    private final ShopGoodsInfoService goodsInfoService;
    private final RedisCache redisCache;
    private final TencentCloudIm tencentCloudIm;
    private final LiveWatchRecordService liveWatchRecordService;
    private final ShopOrderSkuService shopOrderSkuService;
    private final AnchorLivesGiftRecordService livesGiftRecordService;
    private final LiveRoomGiftBusiness liveRoomGiftBusiness;
    private final LivesRoomAdminService livesRoomAdminService;
    private final ApplicationContext applicationContext;
    private final ShopGoodsSkuMapper shopGoodsSkuMapper;
    private final ShopGoodsSkuStocksMapper shopGoodsSkuStocksMapper;


    /**
     * 根据任务编号开播
     */
    public void startLive(Long sellTaskId) {
        //1.查询到主播
        H2Anchor anchor = anchorMapper.queryAnchorByUserId(SecurityUtils.getUserId());
        if (null == anchor) {
            throw new QysException(AnchorMessageEnum.请先申请主播);
        }
        //2.查询到主播带货任务
        AnchorSellTask sellTask = sellTaskService.getById(sellTaskId);
        if (null == sellTask) {
            throw new QysException("请检查ID");
        }
        //3.判断状态
        AnchorSellTaskStatus.checkStatus(sellTask.getStatus(), AnchorSellTaskStatus.待开播);
        //4.判断申请带货直播
        AnchorSellLivesApply sellLivesApply = sellLivesApplyService.checkRepeat(sellTaskId, anchor.getId());
        //5.未申请直播
        if (sellLivesApply == null) {
            throw new QysException(MessageUtils.message("主播带货任务未申请开播"));
        }
        //查询主播
        //if(Objects.equals())

        //1:

    }

    /**
     * 获取推流地址
     */
    @Transactional(rollbackFor = Exception.class)
    public String getPushUrl(Long taskId) {

        Long userId = SecurityUtils.getUserId();

        AnchorLivesRoom room = livesRoomService.getRoomByUserId(userId);
        if(room == null){
            //非主播请求直接返回
            return null;
        }

        if (room.getLiveStatus() == (byte) 2) {
            String streamTaskId = redisCache.getCacheObject(RedisConstant.SUSPEND_LIVE + userId);
            if(StringUtils.isNotEmpty(streamTaskId)){
                //删除拉流任务
                LiveStreamUtils.exampleDeleteAPullTask(streamTaskId);
                redisCache.deleteObject(RedisConstant.SUSPEND_LIVE + userId);
            }
            //状态处于直播中
            return LiveStreamUtils.getPushUrl(userId);
        }

        //查询到主播带货任务
        AnchorSellTask sellTask = sellTaskService.getById(taskId);
        if (null == sellTask || sellTask.getStatus() >= (byte) 2) {
            throw new QysException("no SellTask or SellTask over");
        }

        //判断申请带货直播申请和状态
        AnchorSellLivesApply sellLivesApply = sellLivesApplyService.checkRepeat(taskId, room.getAnchorId());
        if (sellLivesApply == null || sellLivesApply.getPassFlag() != (byte) 2) {
            throw new QysException(MessageUtils.message("主播带货任务未申请开播"));
        }


        //更新直播间状态
        AnchorLivesRoom livesRoom = new AnchorLivesRoom();
        livesRoom.setId(room.getId());
        livesRoom.setLiveStatus((byte) 2);
        livesRoom.setLastLiveTime(LocalDateTime.now());
        livesRoomService.updateById(livesRoom);


        AnchorLivesSession one = livesSessionService.getOne(Wrappers.<AnchorLivesSession>lambdaQuery().eq(AnchorLivesSession::getAnchorSellTaskId, taskId).eq(AnchorLivesSession::getLivesStatus, (byte) 1));
        if (null == one) {
            //创建直播场次（实际已不会涉及到该部分逻辑，因为审批通过直播申请后已经创建了直播场次）
            AnchorLivesSession livesSession = new AnchorLivesSession();
            livesSession.setLivesRoomId(room.getId());
            livesSession.setAnchorId(room.getAnchorId());
            livesSession.setAnchorName(room.getAnchorName());
            livesSession.setLivesStatus((byte) 2);
            livesSession.setStartTime(LocalDateTime.now());
            livesSession.setAnchorSellTaskId(taskId);
            livesSession.setUpdateTime(LocalDateTime.now());
            livesSession.setUpdateBy(SecurityUtils.getUsername());
            livesSessionService.save(livesSession);
        } else {
            //由于断流更新场次状态为直播中
            one.setLivesStatus((byte)2);
            if(one.getStartTime() == null){
                one.setStartTime(LocalDateTime.now());
            }
            livesSessionService.updateById(one);
        }


        //更新带货任务状态
        AnchorSellTask anchorSellTask = new AnchorSellTask();
        anchorSellTask.setId(sellTask.getId());
        anchorSellTask.setStatus((byte) 1);
        sellTaskService.updateById(anchorSellTask);

        List<Long> goodsId = shopSellTaskProductService.listGoodsInfoIdBySellTaskId(anchorSellTask.getShopSellTaskId());
        //更新商品的直播状态
        if (!CollectionUtils.isEmpty(goodsId)) {
            shopGoodsInfoMapper.changeGoodsInfoLiveStatus(goodsId,1);
        }
        H2Anchor h2Anchor = anchorMapper.selectById(room.getAnchorId());
        h2Anchor.setLiveStatus(1);
        anchorMapper.updateById(h2Anchor);

        return LiveStreamUtils.getPushUrl(userId);
    }

    /**
     * 获取播放地址
     */
    public LivesStreamUrlVO getPlayUrl(Long userId, Long excludeUserId) {

        LivesStreamUrlVO vo = new LivesStreamUrlVO();

        if(excludeUserId != null){
            //随机获取直播间
            Set<Object> cacheSet = redisCache.getCacheSet(RedisConstant.LIVE_RANDOM_NUMBER + SecurityUtils.getUserId());
            if(CollectionUtils.isEmpty(cacheSet)){
                cacheSet = Sets.newHashSet(excludeUserId);
            }
            H2Anchor h2Anchor = anchorMapper.selectOne(Wrappers.<H2Anchor>lambdaQuery().eq(H2Anchor::getLiveStatus, 1)
                    .notIn(H2Anchor::getUserId, cacheSet)
                    .last("limit 1"));
            if(h2Anchor == null){
                return null;
            }
            userId = h2Anchor.getUserId();
            cacheSet.add(userId);
            redisCache.setCacheSet(RedisConstant.LIVE_RANDOM_NUMBER + SecurityUtils.getUserId(), cacheSet);
            vo.setAnchorUserId(String.valueOf(userId));
        } else {
            //重新进入直播间清理缓存数据
            redisCache.deleteObject(RedisConstant.LIVE_RANDOM_NUMBER + SecurityUtils.getUserId());
        }

        //判断用户是否在直播
        AnchorLivesRoom livesRoom = livesRoomService.getRoomByUserId(userId);
        if (livesRoom == null || livesRoom.getLiveStatus() == (byte) 1) {
            return null;
        }
        
        //获取直播场次
        AnchorLivesSession livesSession = livesSessionService.getOne(Wrappers.<AnchorLivesSession>lambdaQuery().eq(AnchorLivesSession::getLivesRoomId, livesRoom.getId()).eq(AnchorLivesSession::getLivesStatus, (byte)2));
        if(livesSession.getAnchorSellTaskId() == null){
            vo.setHasTask(Boolean.FALSE);
        } else {
            vo.setHasTask(Boolean.TRUE);
        }

        //判断是否被禁止进入直播间
        Boolean ban = livesRoomAdminService.isBan(livesSession.getId(), SecurityUtils.getUserId());
        if(ban){
            vo.setCode(BANNED);
            return vo;
        }
        vo.setStreamUrl(LiveStreamUtils.getPlayUrl(userId));
        return vo;
    }

    /**
     * 初始化直播间
     */
    public void initLiveRoom(Long userId) {
        H2Anchor anchor = anchorService.getOne(Wrappers.<H2Anchor>lambdaQuery().eq(H2Anchor::getUserId, userId).eq(H2Anchor::getIsDelete, 0));
        if (anchor == null) {
            throw new QysException("not anchor");
        }
        if(Constants.YesNoInt.YES.equals(anchor.getAnchorStatus())){
            throw new QysException(MessageUtils.message("state.frozen"));
        }
        if(Constants.YesNoInt.NO.equals(anchor.getLiveEntertainment())){
            throw new QysException(MessageUtils.message("no.live.permissions"));
        }
        livesRoomService.initLivesRoom(anchor.getId());
    }


    /**
     * 关闭直播间
     *
     * @param userId       主播userId
     * @param destroyGroup 是否解散IM群
     * @param endTime      指定直播结束时间
     */
    @Transactional(rollbackFor = Exception.class)
    public void closeLive(Long userId, boolean destroyGroup, LocalDateTime endTime) {

        AnchorLivesRoom room = livesRoomService.getRoomByUserId(userId);
        if (room == null || room.getLiveStatus() == (byte) 1) {
            return;
        }
        //更新直播间状态
        AnchorLivesRoom livesRoom = new AnchorLivesRoom();
        livesRoom.setId(room.getId());
        livesRoom.setLiveStatus((byte) 1);
        livesRoomService.updateById(livesRoom);

        //更新直播场次表信息
        AnchorLivesSession livesSession = livesSessionService.getIsStarted(room.getId());
        livesSession.setEndTime(endTime == null ? LocalDateTime.now() : endTime);
        int livesDuration = (int)(System.currentTimeMillis() - localDateTimeTOLong(livesSession.getStartTime()))/(1000 * 60) + 1;
        livesSession.setLivesDuration(livesDuration);
        livesSession.setLivesStatus((byte) 1);
        livesSession.setSourceType(StringUtils.EMPTY);
        livesSession.setAutoPushAddress(StringUtils.EMPTY);
        livesSessionService.updateById(livesSession);

        //有任务关播处理
        if(livesSession.getAnchorSellTaskId() != null){
            AnchorSellTask sellTask = sellTaskService.getById(livesSession.getAnchorSellTaskId());
            if (sellTask != null) {
                //更新主播带货任务状态
                sellTask.setStatus((byte)2);
                sellTaskService.updateById(sellTask);
                //更新商品的直播状态
                List<Long> goodsId = shopSellTaskProductService.listGoodsInfoIdBySellTaskId(sellTask.getShopSellTaskId());
                if (!CollectionUtils.isEmpty(goodsId)) {
                    shopGoodsInfoMapper.changeGoodsInfoLiveStatus(goodsId,0);
                }
            }
        }


        //结算直播收益
        this.liveTaskSettlement(livesSession);

        //更新主播状态
        H2Anchor h2Anchor = anchorMapper.selectById(livesSession.getAnchorId());
        h2Anchor.setLiveStatus(0);
        anchorMapper.updateById(h2Anchor);

        if(destroyGroup){
            //解散IM群
            tencentCloudIm.destroyGroup(room.getLivesRoomNum());
        }

        String taskId = redisCache.getCacheObject(RedisConstant.AUTOMATIC_LIVE + userId);
        if(StringUtils.isNotEmpty(taskId)){
            //如果存在切流任务，删除任务
            LiveStreamUtils.exampleDeleteAPullTask(taskId);
            redisCache.deleteObject(RedisConstant.AUTOMATIC_LIVE + userId);
        }
        String suspendTaskId = redisCache.getCacheObject(RedisConstant.SUSPEND_LIVE + userId);
        if(StringUtils.isNotEmpty(suspendTaskId)){
            //如果存在暂停切流，删除任务
            LiveStreamUtils.exampleDeleteAPullTask(suspendTaskId);
            redisCache.deleteObject(RedisConstant.SUSPEND_LIVE + userId);
        }

        //发布直播结束消息
        LiveEndEventPOJO liveEndEventPOJO = new LiveEndEventPOJO();
        liveEndEventPOJO.setUserId(userId);
        liveEndEventPOJO.setLivesSessionId(livesSession.getId());
        liveEndEventPOJO.setLivesDuration(livesDuration);
        applicationContext.publishEvent(new LiveEndEvent(liveEndEventPOJO));
    }

    /**
     * 结算直播场次
     * @param session 直播场次
     */
    public void liveTaskSettlement(AnchorLivesSession session) {

        AnchorLivesSession livesSession = new AnchorLivesSession();
        //更新状态
        livesSession.setLivesStatus((byte)3);
        //结算观看人数
        Long size = redisCache.size(RedisConstant.LIVE_ONLINE_NUMBER_PREFIX + session.getId());
        livesSession.setViewsNumber(size.intValue());
        //结算销售数据
        Map<String, Object> slesData = this.getSlesData(session.getId());
        livesSession.setOrderQuantity((Integer) slesData.get("orderQuantity"));
        livesSession.setSalesVolume((Integer) slesData.get("salesVolume"));
        livesSession.setSalesAmount((BigDecimal) slesData.get("salesAmount"));
        //结算金币收益
        livesSession.setGoldNum(this.getGoldNum(session.getId()));
        livesSessionService.update(livesSession, Wrappers.<AnchorLivesSession>lambdaUpdate().eq(AnchorLivesSession::getId, session.getId()));

        //结算任务
        AnchorSellTask anchorSellTask = new AnchorSellTask();
        anchorSellTask.setId(session.getAnchorSellTaskId());
        anchorSellTask.setStatus((byte)2);
        sellTaskService.update(anchorSellTask, Wrappers.<AnchorSellTask>lambdaUpdate().eq(AnchorSellTask::getId, session.getAnchorSellTaskId()));

        //清空当前直播场次数据
        redisCache.deleteObject(RedisConstant.LIVE_ONLINE_NUMBER_PREFIX + session.getId());
        //清空当前直播场次礼物数据
        redisCache.deleteObject(RedisConstant.LIVE_ROOM_GIFT_LIST + session.getId());
    }

    private Map<String, Object> getSlesData(Long livesSessionId){

        Map<String, Object> resultMap = Maps.newHashMap();
        resultMap.put("orderQuantity", 0);
        resultMap.put("salesVolume", 0);
        resultMap.put("salesAmount", BigDecimal.ZERO);

        LambdaQueryWrapper<ShopOrderSku> queryWrapper = Wrappers.<ShopOrderSku>lambdaQuery()
                .eq(ShopOrderSku::getTaskId, livesSessionId)
                .isNotNull(ShopOrderSku::getOrderStatus);
        //查询当前场次当前主播的销售数据
        List<ShopOrderSku> slesDataList = shopOrderSkuService.list(queryWrapper);

        if (!CollectionUtils.isEmpty(slesDataList)) {
            //订单量
            resultMap.put("orderQuantity", slesDataList.size());
            //销量
            resultMap.put("salesVolume", slesDataList.stream().mapToInt(ShopOrderSku::getNum).sum());
            //销售额
            resultMap.put("salesAmount", slesDataList.stream().map(ShopOrderSku::getMoney).reduce(BigDecimal.ZERO, BigDecimal::add));
        }
        return resultMap;
    }


    private BigDecimal getGoldNum(Long livesSessionId){
        Wrapper<AnchorLivesGiftRecord> queryWrapper = Wrappers.<AnchorLivesGiftRecord>lambdaQuery()
                .select(AnchorLivesGiftRecord::getId, AnchorLivesGiftRecord::getQuantity, AnchorLivesGiftRecord::getGiftGold)
                .eq(AnchorLivesGiftRecord::getLivesSessionId, livesSessionId);

        List<AnchorLivesGiftRecord> giftRecords = livesGiftRecordService.list(queryWrapper);
        if(CollectionUtils.isEmpty(giftRecords)){
            return BigDecimal.ZERO;
        }
        //金币兑换比例
        BigDecimal goldRatio = liveRoomGiftBusiness.exchangeGoldRatio();
        //礼物总金币价值
        return giftRecords.stream().map(e -> e.getGiftGold().multiply(BigDecimal.valueOf(e.getQuantity()))).reduce(BigDecimal.ZERO, BigDecimal::add).divide(goldRatio,2, RoundingMode.DOWN);
    }

    /**
     * 获取直播间信编号
     *
     * @param userId
     * @return
     */
    public String getLivesRoomNum(String userId) {
        AnchorLivesRoom roomByUserId = livesRoomService.getRoomByUserId(Long.valueOf(userId));
        if (roomByUserId != null) {
            return roomByUserId.getLivesRoomNum();
        }
        return null;
    }


    /**
     * 获取直播间场次
     *
     * @param userId
     * @return
     */
    public String getLivesSession(String userId) {
        AnchorLivesRoom livesRoom = livesRoomService.getRoomByUserId(Long.valueOf(userId));
        if (livesRoom != null) {
            AnchorLivesSession session = livesSessionService.getIsStarted(livesRoom.getId());
            if (session != null) {
                return String.valueOf(session.getId());
            }
        }
        return null;
    }

    /**
     * 更新主播离开时间（断流）
     *
     * @param userId 用户id
     * @param clear  是否清除（恢复直播则清除）
     * @return
     */
    public void updateDepartureTime(String userId, boolean clear) {

        //延迟执行，保证正常关闭直播间的数据LivesStatus先行变更为结束
        try {
            Thread.sleep(500);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        AnchorLivesRoom livesRoom = livesRoomService.getRoomByUserId(Long.valueOf(userId));
        if (livesRoom != null) {
            AnchorLivesSession session = livesSessionService.getIsStarted(livesRoom.getId());
            if (session != null && session.getLivesStatus() == (byte)2) {
                //1、更新离开时间
                AnchorLivesSession livesSession = new AnchorLivesSession();
                livesSession.setId(session.getId());
                livesSession.setDepartureTime(clear ? null : LocalDateTime.now());
                livesSessionService.updateById(livesSession);

                //2、切流主播离开界面
                this.exitDisconnect(userId);
            }
        }
    }

    /**
     * 根据直播场次查询主播用户id
     *
     * @param livesSessionId 直播场次id
     * @return 用户id
     */
    public String getUserIdByLivesSessionId(Long livesSessionId) {
        AnchorLivesSession session = livesSessionService.getById(livesSessionId);
        if (session != null) {
            H2AppUser user = livesRoomService.getUserByAnchorId(session.getAnchorId());
            if (user != null) {
                return user.getId();
            }
        }
        return null;
    }


    private long localDateTimeTOLong(LocalDateTime time) {
        ZoneId zoneId = ZoneId.systemDefault();
        Instant instant = time.atZone(zoneId).toInstant();
        return instant.toEpochMilli();
    }


    public List<SellTaskGoodsChild> getLiveRoomGoods(String roomNum) {

        //把livesSessionId转换成roomNum(兼容sessionId查询)
        if(roomNum.length() != 8){
            AnchorLivesSession session = livesSessionService.getById(roomNum);
            if(session == null){
                throw new QysException("sessionId error");
            }
            AnchorLivesRoom livesRoom = livesRoomService.queryByAnchorId(session.getAnchorId());
            roomNum = livesRoom.getLivesRoomNum();
        }

        //获取直播场次
        AnchorLivesRoom room = livesRoomService.getOne(Wrappers.<AnchorLivesRoom>lambdaQuery().eq(AnchorLivesRoom::getLivesRoomNum, roomNum));
        if (room == null || room.getLiveStatus() == 1) {
            return null;
        }
        AnchorLivesSession livesSession = livesSessionService.getIsStarted(room.getId());
        if(livesSession.getAnchorSellTaskId() == null){
            return null;
        }
        //根据任务id获取商品id
        AnchorSellTask anchorSellTask = sellTaskService.getById(livesSession.getAnchorSellTaskId());
        List<Long> goodsId = shopSellTaskProductService.listGoodsInfoIdBySellTaskId(anchorSellTask.getShopSellTaskId());
        //查询商品详情
        if (!CollectionUtils.isEmpty(goodsId)) {
            List<ShopGoodsInfo> list = goodsInfoService.list(Wrappers.<ShopGoodsInfo>lambdaQuery().in(ShopGoodsInfo::getId, goodsId));
            List<SellTaskGoodsChild> children = BeanUtil.copyToList(list, SellTaskGoodsChild.class);
            for (SellTaskGoodsChild child : children) {
                //是否已被展示
                Long goodsIdByCache = redisCache.getCacheObject(RedisConstant.LIVE_GOODS_SHOW + livesSession.getId());
                if(goodsIdByCache != null){
                    child.setShow(child.getId().equals(goodsIdByCache) ? 1 : 0);
                } else {
                    child.setShow(0);
                }
                //心愿状态
                String wishByCache = redisCache.getCacheObject(RedisConstant.LIVE_GOODS_WISH + livesSession.getId());
                if(StringUtils.isNotEmpty(wishByCache)){
                    Long id = Long.valueOf(wishByCache.split(":")[0]);
                    if(id.equals(child.getId())){
                        child.setWishStatus(1);
                        child.setWishNumber(Integer.valueOf(wishByCache.split(":")[1]));
                    } else {
                        child.setWishStatus(0);
                    }
                }
                //库存
                List<ShopGoodsSku> shopGoodsSkus = shopGoodsSkuMapper.selectList(Wrappers.<ShopGoodsSku>lambdaQuery().eq(ShopGoodsSku::getShopGoodsId, child.getId()));
                if(!CollectionUtils.isEmpty(shopGoodsSkus)){
                    List<Long> skuId = shopGoodsSkus.stream().map(ShopGoodsSku::getId).toList();
                    ShopGoodsSkuStocks one = shopGoodsSkuStocksMapper.selectOne(Wrappers.<ShopGoodsSkuStocks>query().select("sum(num) as num")
                            .in("sku_id", skuId)
                            .eq("state", 0));
                    if(one != null){
                        child.setInventory(one.getNum());
                    }
                }
            }
            //按照id升序排序，方便前端标记商品编号
            List<SellTaskGoodsChild> collect = children.stream().sorted(Comparator.comparingLong(SellTaskGoodsChild::getId)).toList();
            Map<String, Integer> map = Maps.newHashMap();
            int num = 1;
            for (SellTaskGoodsChild sellTaskGoodsChild : collect) {
                map.put(String.valueOf(sellTaskGoodsChild.getId()), num++);
            }
            redisCache.setCacheMap(RedisConstant.LIVE_GOODS_NUMBER + livesSession.getId(), map, 60L, TimeUnit.MINUTES);
            return collect;
        }
        return null;
    }

    /**
     * 拉流转推
     * @param taskId 带货任务ID
     * @param dto
     * @param openLives 拉流转推请求失败是否重新开启直播间
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public String autoLive(Long taskId, CreatePullFlowDTO dto, boolean openLives) {

        //查询到主播带货任务
        AnchorSellTask sellTask = sellTaskService.getById(taskId);
        if (null == sellTask) {
            throw new QysException("no SellTask by taskId");
        }

        //查询直播间
        AnchorLivesRoom room = livesRoomService.queryByAnchorId(sellTask.getAnchorId());
        if (room == null || room.getLiveStatus() == (byte) 2) {
            throw new QysException("room is liveing");
        }

        //执行拉流转推
        Long userId = anchorService.getById(room.getAnchorId()).getUserId();
        dto.setStreamName(String.valueOf(userId));
        String result = null;
        try {
            result = LiveStreamUtils.createPullFlow(dto);
            if(StringUtils.isBlank(result) && !openLives){
                return result;
            }
        } catch (Exception e) {
            log.error("autoLive error:{}", e.getMessage());
            if(!openLives){
                return result;
            }
        }

        //判断申请带货直播申请和状态
        AnchorSellLivesApply sellLivesApply = sellLivesApplyService.checkRepeat(taskId, room.getAnchorId());
        if (sellLivesApply == null || sellLivesApply.getPassFlag() != (byte) 2) {
            throw new QysException(MessageUtils.message("主播带货任务未申请开播"));
        }

        //更新直播间状态
        AnchorLivesRoom livesRoom = new AnchorLivesRoom();
        livesRoom.setId(room.getId());
        livesRoom.setLiveStatus((byte) 2);
        livesRoom.setLastLiveTime(LocalDateTime.now());
        livesRoomService.updateById(livesRoom);

        AnchorLivesSession one = livesSessionService.getOne(Wrappers.<AnchorLivesSession>lambdaQuery().eq(AnchorLivesSession::getAnchorSellTaskId, taskId).eq(AnchorLivesSession::getLivesStatus, (byte) 1));
        if (null == one) {
            //创建直播场次
            AnchorLivesSession livesSession = new AnchorLivesSession();
            livesSession.setLivesRoomId(room.getId());
            livesSession.setAnchorId(room.getAnchorId());
            livesSession.setAnchorName(room.getAnchorName());
            livesSession.setLivesStatus((byte) 2);
            livesSession.setStartTime(LocalDateTime.now());
            livesSession.setAnchorSellTaskId(taskId);
            livesSessionService.save(livesSession);
        } else {
            //由于断流更新场次状态为直播中
            one.setLivesStatus((byte)2);
            if(one.getStartTime() == null){
                one.setStartTime(LocalDateTime.now());
            }
            livesSessionService.updateById(one);
        }

        //更新带货任务状态
        AnchorSellTask anchorSellTask = new AnchorSellTask();
        anchorSellTask.setId(sellTask.getId());
        anchorSellTask.setStatus((byte) 1);
        sellTaskService.updateById(anchorSellTask);

        //更新商品的直播状态
        List<Long> goodsId = shopSellTaskProductService.listGoodsInfoIdBySellTaskId(anchorSellTask.getShopSellTaskId());
        if (!CollectionUtils.isEmpty(goodsId)) {
            shopGoodsInfoMapper.changeGoodsInfoLiveStatus(goodsId,1);
        }
        H2Anchor h2Anchor = anchorMapper.selectById(room.getAnchorId());
        h2Anchor.setLiveStatus(1);
        anchorMapper.updateById(h2Anchor);

        return result;
    }

    public Boolean startAutoLive(Long id) {
        AnchorLivesSession livesSession = livesSessionService.getById(id);
        if(StringUtils.isNotBlank(livesSession.getAutoPushAddress())){
            if("close".equals(livesSession.getAutoPushAddress())){
                //强制关闭直播使用
                AnchorLivesSession updateLivesSession = new AnchorLivesSession();
                updateLivesSession.setId(livesSession.getId());
                updateLivesSession.setAutoPushAddress(StringUtils.EMPTY);
                livesSessionService.updateById(updateLivesSession);
                return true;
            }
            CreatePullFlowDTO createPullFlowDTO = new CreatePullFlowDTO();
            createPullFlowDTO.setSourceType(livesSession.getSourceType());
            String [] url = {livesSession.getAutoPushAddress()};
            createPullFlowDTO.setSourceUrls(url);
            createPullFlowDTO.setStartTime(System.currentTimeMillis());
            createPullFlowDTO.setEndTime(System.currentTimeMillis() + 8000 * 60 * 60);
            createPullFlowDTO.setStreamName(livesRoomService.getUserByAnchorId(livesSession.getAnchorId()).getId());
            try {
                LiveStreamUtils.createPullFlow(createPullFlowDTO);
            } catch (Exception e) {
                log.error("切流失败:{}", e.getMessage());
            }
            //清理数据
            AnchorLivesSession updateLivesSession = new AnchorLivesSession();
            updateLivesSession.setId(livesSession.getId());
            updateLivesSession.setAutoPushAddress(StringUtils.EMPTY);
            updateLivesSession.setSourceType(StringUtils.EMPTY);
            livesSessionService.updateById(updateLivesSession);
            return true;
        } else {
            return false;
        }

    }

    /**
     * 无任务开播
     * @return 推流地址
     */
    public String getPushUrlNoTask() {

        Long userId = SecurityUtils.getUserId();

        AnchorLivesRoom room = livesRoomService.getRoomByUserId(userId);
        if(room == null){
            //非主播请求直接返回
            return null;
        }
        //判断直播权限
        H2Anchor anchor = anchorMapper.selectById(room.getAnchorId());
        if(Constants.YesNoInt.NO.equals(anchor.getLiveEntertainment())){
            throw new QysException(MessageUtils.message("no.live.permissions"));
        }

        if (room.getLiveStatus() == (byte) 2) {
            String taskId = redisCache.getCacheObject(RedisConstant.SUSPEND_LIVE + userId);
            if(StringUtils.isNotEmpty(taskId)){
                //删除拉流任务
                LiveStreamUtils.exampleDeleteAPullTask(taskId);
                redisCache.deleteObject(RedisConstant.SUSPEND_LIVE + userId);
            }
            //状态处于直播中
            return LiveStreamUtils.getPushUrl(userId);
        }

        //更新直播间状态
        AnchorLivesRoom livesRoom = new AnchorLivesRoom();
        livesRoom.setId(room.getId());
        livesRoom.setLiveStatus((byte) 2);
        livesRoom.setLastLiveTime(LocalDateTime.now());
        livesRoomService.updateById(livesRoom);

        //创建直播场次
        AnchorLivesSession livesSession = new AnchorLivesSession();
        livesSession.setLivesRoomId(room.getId());
        livesSession.setAnchorId(room.getAnchorId());
        livesSession.setAnchorName(room.getAnchorName());
        livesSession.setLivesStatus((byte) 2);
        livesSession.setStartTime(LocalDateTime.now());
        livesSession.setUpdateTime(LocalDateTime.now());
        livesSession.setUpdateBy(SecurityUtils.getUsername());
        livesSessionService.save(livesSession);

        //更新主播直播状态
        H2Anchor h2Anchor = anchorMapper.selectById(room.getAnchorId());
        h2Anchor.setLiveStatus(1);
        anchorMapper.updateById(h2Anchor);

        return LiveStreamUtils.getPushUrl(userId);
    }

    public Boolean showGoods(Long goodsId) {

        //获取商品详情
        ShopGoodsInfo info = goodsInfoService.getById(goodsId);
        if(info == null){
            throw new QysException("goodsId error");
        }

        //判断直播间状态
        AnchorLivesRoom room = livesRoomService.getRoomByUserId(SecurityUtils.getUserId());
        if(room == null){
            throw new QysException("no lives room");
        }
        if(room.getLiveStatus() != (byte)2 ){
            throw new QysException("no living");
        }

        //获取直播场次(记录商品展示信息)
        AnchorLivesSession livesSession = livesSessionService.getIsStarted(room.getId());

        //是否已指定展示商品
        Long goodsIdByCache = redisCache.getCacheObject(RedisConstant.LIVE_GOODS_SHOW + livesSession.getId());

        HashMap<String, String> map = Maps.newHashMap();
        map.put("goodsId", StringUtils.EMPTY);
        map.put("goodsName", StringUtils.EMPTY);
        map.put("goodsImage", StringUtils.EMPTY);
        map.put("goodsPrice", StringUtils.EMPTY);

        if(goodsIdByCache != null && goodsIdByCache.equals(goodsId)){
            Boolean res = tencentCloudIm.modifyGroupInfo(room.getLivesRoomNum(), map ,true);
            if(res){
                redisCache.deleteObject(RedisConstant.LIVE_GOODS_SHOW + livesSession.getId());
            }
            return res;
        } else {
            //发送到im
            map.put("goodsId", String.valueOf(info.getId()));
            map.put("goodsName", info.getGoodsName());
            map.put("goodsImage", info.getImage());
            map.put("goodsPrice", String.valueOf(info.getPrice()));
            Boolean res = tencentCloudIm.modifyGroupInfo(room.getLivesRoomNum(), map, true);
            if(res){
                redisCache.setCacheObject(RedisConstant.LIVE_GOODS_SHOW + livesSession.getId(), goodsId, 60L, TimeUnit.MINUTES);
            }
            return res;
        }
    }

    public Boolean setWish(Long goodsId, Integer total) {
        //获取商品详情
        ShopGoodsInfo info = goodsInfoService.getById(goodsId);
        if(info == null){
            throw new QysException("goodsId error");
        }

        //判断直播间状态
        AnchorLivesRoom room = livesRoomService.getRoomByUserId(SecurityUtils.getUserId());
        if(room == null){
            throw new QysException("no lives room");
        }
        if(room.getLiveStatus() != (byte)2 ){
            throw new QysException("no living");
        }

        //获取直播场次(记录商品展示信息)
        AnchorLivesSession livesSession = livesSessionService.getIsStarted(room.getId());

        //是否已指定展示商品,格式为 goodsId:total:finish
        String wishByCache = redisCache.getCacheObject(RedisConstant.LIVE_GOODS_WISH + livesSession.getId());

        HashMap<String, String> map = Maps.newHashMap();
        map.put("wishId", StringUtils.EMPTY);
//        map.put("wishName", StringUtils.EMPTY);
        map.put("wishImage", StringUtils.EMPTY);
//        map.put("wishPrice", StringUtils.EMPTY);
        map.put("wishTotal", StringUtils.EMPTY);
        map.put("wishFinish", StringUtils.EMPTY);

        if(StringUtils.isNotEmpty(wishByCache) && Long.valueOf(wishByCache.split(":")[0]).equals(goodsId) && Integer.valueOf(wishByCache.split(":")[1]).equals(total)){
            Boolean res = tencentCloudIm.modifyGroupInfo(room.getLivesRoomNum(), map, true);
            if(res){
                redisCache.deleteObject(RedisConstant.LIVE_GOODS_WISH + livesSession.getId());
            }
            return res;
        } else {
            //发送到im
            map.put("wishId", String.valueOf(info.getId()));
//            map.put("wishName", info.getGoodsName());
            map.put("wishImage", info.getImage());
//            map.put("wishPrice", String.valueOf(info.getPrice()));
            map.put("wishTotal", String.valueOf(total));
            map.put("wishFinish", "0");
            Boolean res = tencentCloudIm.modifyGroupInfo(room.getLivesRoomNum(), map, true);
            if(res){
                redisCache.setCacheObject(RedisConstant.LIVE_GOODS_WISH + livesSession.getId(), goodsId + ":" + total + ":0", 60L, TimeUnit.MINUTES);
            }
            return res;
        }
    }

    public SettlementDataVO settlementData(Long livesSessionId) {

        AnchorLivesSession session = livesSessionService.getById(livesSessionId);
        if(session == null){
            return null;
        }

        //获取本场主播的用户信息
        H2AppUser user = livesRoomService.getUserByAnchorId(session.getAnchorId());

        SettlementDataVO settlementDataVO = new SettlementDataVO();

        AnchorLivesRoom room = livesRoomService.getRoomByUserId(SecurityUtils.getUserId());
        //判断请求用户是本场主播还是观众
        if(room !=null && room.getAnchorId().equals(session.getAnchorId())){
            BeanUtil.copyProperties(session, settlementDataVO);

        } else {
            settlementDataVO.setViewsNumber(session.getViewsNumber());
        }
        settlementDataVO.setAnchorAvatar(user.getAvatar());
        settlementDataVO.setAnchorName(user.getNickName());
        return settlementDataVO;
    }

    /**
     * 根据用户获取直播间编号和场次id
     * @param userId 主播的userId
     * @return
     */
    public LiveInfoVO getLiveRoomInfo(String userId) {
        LiveInfoVO liveInfoVO = new LiveInfoVO();
        liveInfoVO.setLivesSessionId(this.getLivesSession(userId));
        liveInfoVO.setLivesRoomNum(this.getLivesRoomNum(userId));

        //非主播操作记录
        if(!Objects.equals(SecurityUtils.getUserId(), Long.valueOf(userId))){
            //统计观看人数
            String watchUser = String.valueOf(SecurityUtils.getUserId());
            String key = RedisConstant.LIVE_ONLINE_NUMBER_PREFIX + liveInfoVO.getLivesSessionId();
            Long size = redisCache.size(key);
            Set<String> cacheSet;
            if(size == 0){
                cacheSet = Sets.newHashSet(watchUser);
            } else {
                cacheSet = redisCache.getCacheSet(key);
                cacheSet.add(watchUser);
            }
            redisCache.setCacheSet(key, cacheSet);
            //观看记录
            AnchorLivesRoom livesRoom = livesRoomService.getRoomByUserId(Long.valueOf(userId));
            LiveWatchRecord liveWatchRecord = new LiveWatchRecord();
            liveWatchRecord.setUserId(SecurityUtils.getUserId());
            liveWatchRecord.setLivesRoomId(livesRoom.getId());
            liveWatchRecordService.save(liveWatchRecord);
        }
        return liveInfoVO;
    }

    public Map<String, Long> checkLiveStatus(Long userId) {
        AnchorLivesRoom room = livesRoomService.getRoomByUserId(userId);
        if(null == room){
            return null;
        }
        AnchorLivesSession livesSession = livesSessionService.getIsStarted(room.getId());
        if(null == livesSession){
            return null;
        }
        Map<String, Long> map = Maps.newHashMap();
        map.put("taskId", livesSession.getAnchorSellTaskId());
        return map;
    }

    public boolean exitDisconnect(String userId) {

        String taskId = redisCache.getCacheObject(RedisConstant.SUSPEND_LIVE + userId);
        if (StringUtils.isNotEmpty(taskId)) {
            //删除拉流任务
            LiveStreamUtils.exampleDeleteAPullTask(taskId);
            redisCache.deleteObject(RedisConstant.SUSPEND_LIVE + userId);
        }

        try {
            CreatePullFlowDTO dto = new CreatePullFlowDTO();
            dto.setSourceType("PullVodPushLive");
            //注意：更换离开垫片请同步处理忽略垫片回调
            String [] url = {"https://live-mall-1319900789.cos.ap-singapore.myqcloud.com/short_video%2F1703042859730.mp4"};
            dto.setSourceUrls(url);
            dto.setStartTime(System.currentTimeMillis());
            dto.setEndTime(System.currentTimeMillis() + 1000 * 60 * 30);
            dto.setStreamName(userId);
            taskId = LiveStreamUtils.createPullFlow(dto);
            if(StringUtils.isNotEmpty(taskId)){
                redisCache.setCacheObject(RedisConstant.SUSPEND_LIVE + userId, taskId, 60L, TimeUnit.MINUTES);
            } else {
                return false;
            }
        } catch (Exception e) {
            log.error("暂停直播失败:{}", e.getMessage());
            return false;
        }

        //在创建拉流转推任务后延迟3秒切备用流
        ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(1);
        threadPool.schedule(()-> LiveStreamUtils.switchBackupStream(userId), 1, TimeUnit.SECONDS);
        threadPool.shutdown();
        return true;
    }

    /**
     * 重新推流（恢复直播）
     * @param userId
     */
    public void repushFlow(String userId) {
        String taskId = redisCache.getCacheObject(RedisConstant.SUSPEND_LIVE + userId);
        if (StringUtils.isNotEmpty(taskId)) {
            //删除拉流任务
            LiveStreamUtils.exampleDeleteAPullTask(taskId);
            redisCache.deleteObject(RedisConstant.SUSPEND_LIVE + userId);
        }
    }
}
